package momo.app.status;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.Vector;

import momo.app.comms.CommsHistory;
import momo.app.comms.CommsManager;
import momo.app.config.AppConfig;
import momo.app.structure.Message;
import momo.app.structure.Peer;
import momo.app.structure.PeerLink;
import momo.app.structure.PeerList;
import momo.app.structure.SessionController;
import momo.app.structure.StabMap;
import momo.app.util.FileOutput;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class StatusManager
{
	private Log log = LogFactory.getLog(StatusManager.class);
	private Log logStructure = LogFactory.getLog("MoMoStructureLog");
	
	public static final int WORKLOAD_SMOOTH_TRESHOLD = 1000;
	public static final int MIN_SUM_WORKLOAD_TRESHOLD = 10;
	public static final double WORKLOAD_SMOOTH_BASE = 0.8;
	
	private static StatusManager manager;
	private StatusListener listener;
	
	private PeerList peerList;
//	private Peer me;
	
	//private Peer root;
	
	private String rootId;
	
	private HashMap<String, LatencyMeasurement> latMeasureMap;
	
	private StabMap stabMap;
	
	private double score = 0;
	
	private StatusManager()
	{
		
		if ( log.isDebugEnabled() )
			log.debug("creating new StatusManager");
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("creating new StatusManager");
		
		peerList = new PeerList();
		latMeasureMap = new HashMap<String, LatencyMeasurement>();
		setMeAsRoot();
		SessionController.getInstance().initSession();
		addPeer(AppConfig.getInstance().getMe(), "StatusManager const()");
		stabMap = new StabMap();
		this.noteTimeJoined(AppConfig.getInstance().getId());
		displayStabMap("constructor");
		LatencyPoolThread latPool = new LatencyPoolThread();
		
		//test for planetlab
		PlanetLabPeriodicalCheck check = new PlanetLabPeriodicalCheck();
		check.run();
	}
	
	public void setTreeScore(double score) {
		this.score = score;
	}
	
	public double getTreeScore() {
		return score;
	}
	
	public static boolean isStatusManagerCreated() {
		return manager != null;
	}
	
	public synchronized static StatusManager getInstance()
	{
		if ( manager == null )
		{
			manager = new StatusManager();
		}
		return manager;
	}
	
	synchronized public void listen()
	{
		if ( listener == null )
			listener = new StatusListener(AppConfig.getInstance().getStatusPort());
	}
	
	synchronized public void stopListening()
	{
		if ( listener != null )
		{
			listener.stopListening();
			listener = null;
		}
	}
	
	synchronized public void stopAllComms()
	{
		stopListening();
		
		Set<String> keys = latMeasureMap.keySet();
		Iterator<String> iter = keys.iterator();
		while ( iter.hasNext() )
			latMeasureMap.get(iter.next()).stopMeasuring();
		latMeasureMap.clear();
		
		removeAllExceptMe();
	}
	
	synchronized public void addLatMeasureThread(Peer peer)
	{
		if ( !latMeasureMap.containsKey(peer.getId()) )
		{
			LatencyMeasurement latMeasure = new LatencyMeasurement(peer);
			latMeasureMap.put(peer.getId(), latMeasure);
		}
	}
	
	synchronized public boolean removeLatMeasureThread(String id)
	{
		LatencyMeasurement latMeasure = latMeasureMap.get(id);
		if ( latMeasure == null )
			return false;
		else
		{
			latMeasure.stopMeasuring();
			latMeasureMap.remove(id);
			return true;
			
		}
	}
	
	synchronized public boolean addPeer(Peer peer, String who)
	{
		boolean hasAdd = peerList.add(peer);
		if ( hasAdd )
		{
			if ( log.isDebugEnabled() )
				log.debug(who + " --> Peer added: " + peer.toString());
		}
		return hasAdd;
	}
	
	synchronized public boolean addPeerLink(PeerLink link)
	{
		return peerList.addPeerLink(link);
	}
	
	/**
	 * Adds link if not found else updates link info (cost, latency)
	 * @param source_id
	 * @param dest_id
	 * @param cost
	 * @param lat
	 * @return
	 */
	synchronized public boolean addPeerLink(String source_id, String dest_id, double cost, double lat)
	{
		if ( log.isDebugEnabled() )
			log.debug("addPeerLink("+source_id+", "+dest_id+", "+cost+", "+lat+")");
		return peerList.addPeerLink(source_id, dest_id, cost, lat);
	}
	
	synchronized public boolean updatePeerWorkloadRate(String id, double workloadRate)
	{
		return peerList.updatePeerWorkloadRate(id, workloadRate);
	}
	
	synchronized public boolean updatePeerStability(String id, double stability)
	{
		return peerList.updatePeerStability(id, stability);
	}
	
	synchronized public boolean updatePeer(String id, double workload, double stability)
	{
		return peerList.updatePeer(id, workload, stability);
	}
	
	synchronized public boolean updatePeerAddressToUse(String id, InetAddress addressToUse)
	{
		return peerList.updatePeerAddressToUse(id, addressToUse);
	}
	
	synchronized public boolean updatePeerLinkCost(String source_id, String dest_id, double cost)
	{
		return peerList.updatePeerLinkCost(source_id, dest_id, cost);
	}
	
	synchronized public boolean updatePeerLinkLatency(String source_id, String dest_id, double latency)
	{
		return peerList.updatePeerLinkLatency(source_id, dest_id, latency);
	}
	
	synchronized public boolean updatePeerLink(String source_id, String dest_id, double cost, double latency)
	{
		return peerList.updatePeerLink(source_id, dest_id, cost, latency);
	}
	
	synchronized public boolean removePeer(String id)
	{
		boolean hasRem = peerList.removePeer(id);
		if ( hasRem )
		{
			if ( log.isDebugEnabled() )
				log.debug("Peer removed: " + id);
		}
		return hasRem;
	}
	
	synchronized public boolean removePeerLink(String source_id, String dest_id)
	{
		return peerList.removePeerLink(source_id, dest_id);
	}
	
	synchronized public void removeAssociationOnPeer(String id)
	{
//		this will remove its peerlink as well (defined in graph object)
		removePeer(id);
		removeLatMeasureThread(id);
		calWorkload("removeAssociationOnPeer");
		rescaleWorkloadRate();
		timeLeave(id);
		timeReconstructExistingPeers();
		synchExistingPeerWithActiveStab();
		displayStabMap("removeAssociationOnPeer");
		
		CommsHistory.getInstance().removeState(id);
	}
	
	synchronized public Set<Peer> getPeers()
	{
		return peerList.getPeers();
	}
	
	synchronized public Set<PeerLink> getPeerLinks()
	{
		return peerList.getPeerLinks();
	}
	
	synchronized public PeerList getPeerList()
	{
		return peerList;
	}
	
	synchronized public boolean isPeerExist(String id)
	{
		return peerList.isPeerExist(id);
	}
	
	synchronized public void setMeAsRoot()
	{
		setRootId(AppConfig.getInstance().getId());
	}
	
	synchronized public void setRootId(String newRootId)
	{
		rootId = new String(newRootId);
		
		if ( AppConfig.getInstance().getId().equals(rootId ) )
		{
			if ( log.isDebugEnabled() )
				log.debug("I AM THE ROOT: " + rootId);
			if ( logStructure.isDebugEnabled() )
				logStructure.debug("I AM THE ROOT: " + rootId);
		}else {
			if ( log.isDebugEnabled() )
				log.debug("ROOT is: " + rootId);
			if ( logStructure.isDebugEnabled() )
				logStructure.debug("ROOT is: " + rootId);
		}
	}
	
	synchronized public String getRootId()
	{
		return new String(rootId);
	}
	
	/**
	 * check whether i m the root
	 * @return true if i m the root
	 */
	synchronized public boolean isRoot()
	{
		String myId = AppConfig.getInstance().getId();
		if ( log.isDebugEnabled() )
			log.debug("isRoot: currentRoot: "+rootId+" myId: "+myId);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("isRoot: currentRoot: "+rootId+" myId: "+myId);
		
		if (myId.equals(rootId))
			return true;
		
		return false;
	}
	
	/**
	 * check whether this id is the root
	 * @return true if this id is the root
	 */
	synchronized public boolean isRoot(String id)
	{
		if ( id.equals(rootId) )
			return true;
		return false;
	}
	
	public void joinRequest(InetAddress ip, int statusPort) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("Join Request to " + ip.getHostAddress()+":" + statusPort);
		joinToNetwork(ip, statusPort);
	}
	
	private void joinToNetwork(InetAddress ip, int statusPort) throws IOException
	{
		int count = 0;
		boolean isSuccess = false;
		
		while ( count < 3 )
		{
			if ( log.isDebugEnabled() )
				log.debug("JoinToNetwork, Tried: " + count);
			Socket sock = new Socket();
			sock.connect(new InetSocketAddress(ip, statusPort), 30000);
			
			BufferedReader in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
			PrintWriter out = new PrintWriter(sock.getOutputStream(), true);
			
			long start = System.currentTimeMillis();
			
			if ( this.isOnlyOneInNetwork() )
				AppConfig.getInstance().setWorkloadRate(0);
			
			SessionController.getInstance().initSession();
			this.noteTimeJoined(AppConfig.getInstance().getId());
			
			Message msg = new Message(Message.NO_FORWARD, AppConfig.getInstance().getId(), Message.JOIN, AppConfig.getInstance().getMe().toString());
			out.println(msg.toString());
			out.flush();
			
			String data = in.readLine();
			Message rcvMsg = Message.createMessage(data);
			if ( rcvMsg.getMsgType().equals(Message.RESET_STATE_IN_PROGRESS) )
			{
				try
				{
					Thread.sleep(5 * 1000);
				}catch (Exception e) {
					if ( log.isErrorEnabled() )
						log.error("Exception", e);
				}
			}else if ( rcvMsg.getMsgType().equals(Message.JOIN_REPLY) )
			{
				if ( log.isDebugEnabled() ) {
					log.debug("JOIN_REPLY");
					log.debug(rcvMsg.toString());
				}
				
				StringTokenizer tokens = new StringTokenizer(rcvMsg.getMsg(), "|");
				Peer parent = Peer.createPeer(tokens.nextToken());
				parent.setAddressToUse(ip);
				String newRootId = tokens.nextToken();
				this.setRootId(newRootId);
				int sessionId = Integer.parseInt(tokens.nextToken());
				SessionController.getInstance().setSessionId(sessionId);
				
				long end = System.currentTimeMillis();
				double lat = LatencyMeasurement.calculateMeasuredLatency(start, end);
				
				double costFromParentToMe = parent.getUserCost(AppConfig.getInstance().getId());
				double costFrmMeToParent = AppConfig.getInstance().getUserCost(parent.getId());
				double cost = costFromParentToMe + costFrmMeToParent;
				
				out.println(new Message(Message.NO_FORWARD, AppConfig.getInstance().getId(), Message.MISC, lat+","+cost).toString());
				out.flush();
				
				//Time to linkup to my parent via CommsChannel
				PeerLink link = new PeerLink(parent.getId()+":"+AppConfig.getInstance().getMe().getId(),
												parent, 
												AppConfig.getInstance().getMe(),
												cost,
												lat);
				
				CommsManager.getInstance().linkup(parent, link);
				
				isSuccess = true;
				
				if ( logStructure.isDebugEnabled() )
					logStructure.debug("almost ending join; ROOT: "+rootId);
				
			}else
			{
				throw new IOException("Unknown reply");
			}
			
			out.close();
			in.close();
			sock.close();
			
			if ( isSuccess )
				break;
			
			count ++;
		}
		
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("ending join; ROOT: "+rootId);
		if ( !isSuccess )
			throw new IOException("Failure to join");
	}
	
	public void getStabMapInfo(InetAddress ip, int statusPortToConnect) throws IOException
	{
		Socket sock = new Socket();
		sock.connect(new InetSocketAddress(ip, statusPortToConnect), 30000);
		
		BufferedReader in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
		PrintWriter out = new PrintWriter(sock.getOutputStream(), true);
		
		out.println(
				new Message(Message.NO_FORWARD, 
						AppConfig.getInstance().getId(), 
						Message.STAB_MAP_REQUEST, 
					"").toString()
				);
		out.flush();
		
		String data = in.readLine();
		Message msg = Message.createMessage(data);
		this.updateStabMap(msg.getMsg());
		
		displayStabMap("getStabMapInfo");
		
		out.close();
		in.close();
		sock.close();
	}
	
	public void getPeerInfo(InetAddress ip, int statusPortToConnect) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("getPeerInfo of " + ip + ":" + statusPortToConnect);
		
		Socket sock = new Socket();
		sock.connect(new InetSocketAddress(ip, statusPortToConnect), 30000);
		
		BufferedReader in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
		PrintWriter out = new PrintWriter(sock.getOutputStream(), true);
		
		out.println(
				new Message(Message.NO_FORWARD, 
						AppConfig.getInstance().getId(), 
						Message.GET_PEERS, 
					"").toString()
				);
		out.flush();
		
		String data = in.readLine();
		Message msg = Message.createMessage(data);
		
		StringTokenizer tokens = new StringTokenizer(msg.getMsg(), "|");
		while(tokens.hasMoreTokens())
		{
			Peer peer = Peer.createPeer(tokens.nextToken());
			this.addPeer(peer, "getPeerInfo");
		}
		
		out.close();
		in.close();
		sock.close();
	}
	
	public void measureLatency()
	{
		if ( log.isDebugEnabled() )
			log.debug("measure Latency");
		
		Set<Peer> set = peerList.getPeers();
		Iterator<Peer> iter = set.iterator();
		while ( iter.hasNext() )
		{
			Peer p = iter.next();
			if ( p != AppConfig.getInstance().getMe() )
				this.addLatMeasureThread(p);
		}
	}
	
	synchronized public Collection<LatencyMeasurement> getLatMeasureMap()
	{
		return Collections.unmodifiableCollection(latMeasureMap.values());
	}
	
	/**
	 * Routed List will be updated with my id.
	 * Broadcast this message to everyone except
	 * 1.) Myself
	 * 2.) Routed Recipients
	 * @param msg
	 */
	synchronized public void broadcast(Message msg)
	{
		msg.addToRoutedList(AppConfig.getInstance().getId());
		
		Iterator<Peer> iter = this.getPeers().iterator();
		while ( iter.hasNext() )
		{
			Peer peer = iter.next();
			if ( !peer.getId().equals(AppConfig.getInstance().getId()) 
					&& !msg.containsRoutedRecipient(peer.getId()))
			{
				SendStatusMsgThread sendThread = new SendStatusMsgThread(peer, msg.toString());
			}
		}
	}
	
	synchronized public void removeAllExceptMe()
	{
		Vector<String> vecID = new Vector<String>();
		Iterator<Peer> iter = this.getPeers().iterator();
		while ( iter.hasNext() )
		{
			String id = iter.next().getId();
			if ( !id.equals(AppConfig.getInstance().getId()) )
				vecID.add(id);
		}
		
		for(String id: vecID)
			this.removePeer(id);
	}
	
	synchronized public void removeUnconnectedPeers()
	{
		if ( log.isDebugEnabled() )
			log.debug("remove Unconnected Peers from graph");
		
		boolean success = peerList.leftConnectedGraph(AppConfig.getInstance().getMe());
		if ( success )
			this.rescaleWorkloadRate();
		
	}
	
	synchronized public boolean isOnlyOneInNetwork()
	{
		return (this.getPeers().size() == 1);
	}
	
	synchronized public void calWorkload(String who)
	{
		StringBuffer buffer = new StringBuffer("\n"+ who + " --> CAL");
		if ( isOnlyOneInNetwork() )
		{
			AppConfig.getInstance().setWorkloadRate(1);
			buffer.append("\nOnlyOne, workloadRate: 1");
		}else
		{
			int total = this.getTotalWorkload();
			if (total < MIN_SUM_WORKLOAD_TRESHOLD) {
				buffer.append("not modyfing; workload size "+total+" smaller than treshold "+ MIN_SUM_WORKLOAD_TRESHOLD);
			}
			else { // non-zero workload
				Iterator<Peer> iterPeer = this.getPeers().iterator();
				while ( iterPeer.hasNext() )
				{
					Peer currPeer = iterPeer.next();
					double pastWorkloadRate = currPeer.getWorkloadRate();				
					int workload = currPeer.getWorkload();
					currPeer.addToPastWorkload(workload);
					double curWorkloadRate = total==0 ? 0 : ((double)workload/(double)total);
					currPeer.setWorkloadRate( calWorkloadRate(pastWorkloadRate, curWorkloadRate, currPeer.getPastWorkload()) );
					currPeer.setWorkload(0);
					
					buffer.append("\nPeer: " + currPeer.getId()+
							"\nworkload: " + workload + 
							"\npast: " + pastWorkloadRate + 
							"\ncur: " + curWorkloadRate + 
							"\nnew: " + currPeer.getWorkloadRate());
				}
			}
		}
		
		buffer.append("\n============================\n");
		if ( log.isDebugEnabled() )
			log.debug(buffer.toString());
	}
	
	synchronized public void rescaleWorkloadRate()
	{
		StringBuffer buffer = new StringBuffer();
		if ( isOnlyOneInNetwork() )
		{
			AppConfig.getInstance().setWorkloadRate(1);
			buffer.append("\nRescale --> OnlyOne, workloadRate: 1");
		}else
		{
			double total = this.getTotalWorkloadRate();
			buffer.append("\nTotal WorkloadRate: " + total);
			Iterator<Peer> iterPeer = this.getPeers().iterator();
			
			//there might have peers that dun do any work
			//added this check
			if ( total == 0 )
			{
				buffer.append("\nTotal is 0. Setting root as 1, rest is 0");
				while ( iterPeer.hasNext() )
				{
					Peer currPeer = iterPeer.next();
					if ( this.isRoot(currPeer.getId() ))
						currPeer.setWorkloadRate(1);
					else
						currPeer.setWorkloadRate(0);
					
					buffer.append("\nRescale--> Peer: " + currPeer.getId()+
							"\nnew: " + currPeer.getWorkloadRate());
				}
			}else
			{
				while ( iterPeer.hasNext() )
				{
					Peer currPeer = iterPeer.next();
					double pastWorkloadRate = currPeer.getWorkloadRate();
					currPeer.setWorkloadRate( pastWorkloadRate / total );
					buffer.append("\nRescale--> Peer: " + currPeer.getId()+
							"\npast: " + pastWorkloadRate + 
							"\nnew: " + currPeer.getWorkloadRate());
				}
			}
		}
		buffer.append("\n============================\n");
		
		if ( log.isDebugEnabled() )
			log.debug(buffer.toString());
	}
	
	synchronized private double calWorkloadRate(double pastRate, double curRate, int pastWorkload)
	{
		double smoothing = WORKLOAD_SMOOTH_BASE;
		
		if (pastWorkload < WORKLOAD_SMOOTH_TRESHOLD) {
			smoothing = smoothing * (1.0 / ( WORKLOAD_SMOOTH_TRESHOLD - pastWorkload )); 
		}
		
		double newWorkloadRate = ( pastRate * smoothing ) + ( curRate * (1.0 - smoothing) );
		if ( log.isDebugEnabled() )
			log.debug("New WorkloadRate: " + newWorkloadRate);
		return newWorkloadRate;
	}
	
	synchronized private int getTotalWorkload()
	{
		int total = 0;
		
		Iterator<Peer> iterPeer = this.getPeers().iterator();
		while ( iterPeer.hasNext() )
			total += iterPeer.next().getWorkload();
		return total;
	}
	
	synchronized private double getTotalWorkloadRate()
	{
		double total = 0;
		
		Iterator<Peer> iterPeer = this.getPeers().iterator();
		while ( iterPeer.hasNext() )
			total += iterPeer.next().getWorkloadRate();
		return total;
	}
	
	synchronized public boolean incrementWorkload(String id)
	{
		if ( log.isDebugEnabled() )
			log.debug("incrementWorkload of node " + id);
		return peerList.incrementPeerWorkload(id);
	}
	
	synchronized public void noteTimeJoined(String id)
	{
		stabMap.noteTimeJoined(id);
	}
	
	synchronized public void timeLeave(String id)
	{
		stabMap.leave(id);
	}
	
	synchronized public void resetDurationCal(String id)
	{
		stabMap.resetDurationCal(id);
	}
	
	synchronized public void timeResetExistingPeers()
	{
		Iterator<Peer> iterPeer = peerList.getPeers().iterator();
		while ( iterPeer.hasNext() )
			stabMap.resetDurationCal( iterPeer.next().getId() );
	}
	
	synchronized public void timeReconstructExistingPeers()
	{
		Iterator<Peer> iterPeer = peerList.getPeers().iterator();
		while ( iterPeer.hasNext() )
			stabMap.reconstuct( iterPeer.next().getId() );
	}
	
	synchronized public String getStabMap()
	{
		return stabMap.toString();
	}
	
	synchronized public void updateStabMap(String contents)
	{
		stabMap.update(contents);
	}
	
	synchronized public void synchExistingPeerWithActiveStab()
	{
		Iterator<Peer> iterPeer = peerList.getPeers().iterator();
		while ( iterPeer.hasNext() )
		{
			Peer p = iterPeer.next();
			p.setStability( stabMap.getActiveStability(p.getId()) );
		}
	}
	
	synchronized public void displayStabMap(String who)
	{
		StringBuffer buffer = new StringBuffer("\n----------------------------");
		buffer.append("\nStabMap--> " + who + "\n");
		buffer.append(stabMap.testDisplayString());
		buffer.append("----------------------------\n");
		
		if ( log.isDebugEnabled() )
			log.debug(buffer.toString());
	}
	
	/**
	 * Used for testing
	 */
	public void displayPeer()
	{
		StringBuffer buffer = new StringBuffer();
		buffer.append("displayPeer method");
		
		Set<Peer> set = this.getPeers();
		Iterator<Peer> iter = set.iterator();
		while (iter.hasNext())
			buffer.append( "\n" + iter.next().toString() );
		
		if ( log.isDebugEnabled() )
			log.debug(buffer.toString());
	}
	
	/**
	 * Used for testing
	 */
	public void displayPeerLink()
	{
		StringBuffer buffer = new StringBuffer();
		buffer.append("displayPeerLink method");
		
		Set<PeerLink> set = this.getPeerLinks();
		Iterator<PeerLink> iter = set.iterator();
		while (iter.hasNext())
			buffer.append( "\n" + iter.next().toString() );
		
		if ( log.isDebugEnabled() )
			log.debug(buffer.toString());
	}
	
}//end of class StatusManager
